package com.maimao.message.kafka.sender;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author MaoLin Wang
 * @date 2021/1/3 11:13 上午
 */
@Component
public class KafkaSender {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void send(String topic, String key, Object message) {
        try {
            kafkaTemplate.send(topic, key, message);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void send(String topic, Object message) {
        try {
            kafkaTemplate.send(topic, message);
//            kafkaTemplate.send(new ProducerRecord<>(topic,message));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendCallBack(String topic,String key, Object message, ListenableFutureCallback callback) {
        try {
            kafkaTemplate.send(topic,key, message).addCallback(callback);
//            kafkaTemplate.send(new ProducerRecord<>(topic,message));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
